package net.llz.controller;

import net.llz.domain.ProductOrder;
import net.llz.jms.JmsConfig;
import net.llz.jms.PayProducer;
import net.llz.jms.TransactionProducer;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.List;

@RestController
public class PayController {

    @Autowired
    private PayProducer payProducer;

    @Autowired
    private TransactionProducer transactionMQProducer;

    /**
     * 分布式事务投递
     * @param tag
     * @param otherParam
     * @return
     * @throws Exception
     */

    @RequestMapping("/api/v1/pay_cb")
    public Object callback( String tag, String otherParam ) throws Exception {

        Message message = new Message(JmsConfig.TOPIC, tag, tag+"_key",tag.getBytes());

        SendResult sendResult =  transactionMQProducer.getProducer().
                sendMessageInTransaction(message, otherParam);


        System.out.printf("发送结果=%s, sendResult=%s \n", sendResult.getSendStatus(), sendResult.toString());
        return new HashMap<>();
    }

    /**
     * 取模投递
     * @param text
     * @return
     * @throws MQBrokerException
     * @throws RemotingException
     * @throws InterruptedException
     * @throws MQClientException
     */
    @RequestMapping("/api/v1/pay_cb1")
    public Object callback(String text) throws MQBrokerException, RemotingException, InterruptedException, MQClientException {
        Message message = new Message(JmsConfig.TOPIC, "taga", ("hello llz rocketmq=" + text).getBytes());
        //xxx是级别，1表示配置里面的第一个级别，2表示第二个级别
        //"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
        // message.setDelayTimeLevel(3);

        //同步发送
        //SendResult sendResult =payProducer.getProducer().send(message);

        //异步发送
//        payProducer.getProducer().send(message, new SendCallback() {
//            @Override
//            public void onSuccess(SendResult sendResult) {
//                System.out.printf("发送结果=%s, msg=%s ", sendResult.getSendStatus(), sendResult.toString());
//            }
//            @Override
//            public void onException(Throwable e) {
//                e.printStackTrace();
//                //补偿机制，根据业务情况进行使用，看是否进行重试
//            }
//        });


        //选择队列同步发送
//       SendResult sendResult =  payProducer.getProducer().send(message, new MessageQueueSelector() {
//           @Override
//            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
//               int queueNum = Integer.parseInt(arg.toString());
//               return mqs.get(queueNum);
//            }
//
//        }, 3);
//        System.out.printf("发送结果=%s, msg=%s ", sendResult.getSendStatus(), sendResult.toString());

        //选择队列异步发送到指定queue， SendCallback不能用lambda表达式，有两个函数需要被实现
        payProducer.getProducer().send(message, (mqs, msg, args) ->
        {
            int queueNum = Integer.parseInt(args.toString());
            return mqs.get(queueNum);
        }, 3, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.printf("发送结果=%s, msg=%s ", sendResult.getSendStatus(), sendResult.toString());
            }

            @Override
            public void onException(Throwable e) {
                e.printStackTrace();
            }
        });

        return new HashMap<>();

    }

    /**
     * 普通投递
     * @return
     * @throws Exception
     */
    @RequestMapping("/api/v2/pay_cb2")
    public Object callback() throws Exception {
        List<ProductOrder> list = ProductOrder.getOrderList();
        for (int i = 0; i < list.size(); i++) {
            ProductOrder order = list.get(i);
            Message message = new Message(JmsConfig.ORDERLY_TOPIC, "", order.getOrderId() + "", order.toString().getBytes());

            SendResult sendResult = payProducer.getProducer().send(message, new MessageQueueSelector() {
                @Override
                public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
                    Long id = (Long) o;
                    long index = id % list.size();
                    return list.get((int) index);

                }
            }, order.getOrderId());


            System.out.printf("发送结果=%s, sendResult=%s ,orderid=%s, type=%s\n", sendResult.getSendStatus(), sendResult.toString(),order.getOrderId(),order.getType());

        }
        return new HashMap<>();
    }


}
